package cn.tedu.micro.demo;

import cn.tedu.micro.client.RabbitChannelManager;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

@Slf4j
public class RabbitDemo01 {
    //生命一个队列组件
    @Test
    public void declare() throws IOException {
        //1.客户端操作rabbitmq做什么 都需要channel连接支持
        Channel channel = RabbitChannelManager.newChannel();
        //2.声明一个 叫做queue01的队列
        /**
         *String queue: 创建的队列名称,只能存在唯一名称
         *Boolean durable: 队列是否持久化 true在rabbitmq重启之后 队列保持存在 false不存在
         *Boolean exclusive: 队列是绑定当前连接对象(独占 专属) true别的链接无法操作队列 false所有连接都可以操作队列
         *Boolean autoDelete: 队列是否自动删除 true队列从第一个监听消费者开始计算,到最后一个消费者断开连接,自动删除 false不删除
         *Map<String,Object> args: key值和value值 可以说是固定的,能描述队列特性
         *描述声明队列的最大长度:   x-max_length
         *队列存活时间 x-expires: 如果没人用队列,最多可以存在多久毫秒数
         */
        channel.queueDeclare(
                "queue01",
                false,
                false,
                false,
                null);
    }
    @Test
    public void send() throws IOException {
        //1.客户端操作rabbitmq做什么 都需要channel连接支持
        Channel channel = RabbitChannelManager.newChannel();
        //2.组织消息包含的数据内容
        //2.1 body 消息体
        byte[] body="hello rabbitmq".getBytes(StandardCharsets.UTF_8);
        //2.2 消息属性
        AMQP.BasicProperties.Builder builder=new AMQP.BasicProperties.Builder();
        //固定属性
        builder.priority(1000);//消息优先级
        builder.expiration("50000");//消息存储超时毫秒数
        //自定义属性 头
        Map<String,Object> headers=new HashMap<>();
        headers.put("name","王翠花");
        headers.put("age",120);
        builder.headers(headers);
        AMQP.BasicProperties props = builder.build();
        //3.发送出去 发布publish
        /**
         * String exchange: 交换机名称
         * String routingKey: 路由计算key值 入门路由名称 就是队列名称
         * BasicProperties 消息绑定属性
         * byte[] body: 消息体
         */
        channel.basicPublish("","queue01",props,body);
    }
    //测试消费监听
    @Test
    public void consume() throws IOException, InterruptedException {
        //1.拿到客户端对象
        Channel channel = RabbitChannelManager.newChannel();
        //2.消费方法
        //2.1构建消费方法的入参 delivery 回调(messageArrived类似)
        DeliverCallback deliver=new DeliverCallback() {
            /**
             * 方法作用: 只要监听队列 有消息,就会给当前这个消费者推送消息
             * 每条消息,在消费者中handle处理
             * String consumerTag: 消费者身份 rabbitmq分配的表示tag
             * Delivery message: 就是消息对象 发了什么 这里就能收到什么
             */
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //底层消费入口 获取消息 解析消息,调用业务完成消费
                //1.消息体 body
                byte[] body = message.getBody();
                String msg=new String(body,StandardCharsets.UTF_8);
                //2.消息属性
                AMQP.BasicProperties properties = message.getProperties();
                String expiration = properties.getExpiration();
                Integer priority = properties.getPriority();
                Map<String, Object> headers = properties.getHeaders();
                headers.get("name");
                headers.get("age");
                //3.获取消息途径过程 交换机 routingKey
                String exchange = message.getEnvelope().getExchange();
                String routingKey = message.getEnvelope().getRoutingKey();
                log.info("消息:{},从交换机:{},使用路由key:{}达到消费者",msg,exchange,routingKey);
            }
        };
        //2.2构造消费方法的入参 cancel 回调
        CancelCallback cancel=new CancelCallback(){
            /**
             *方法作用: 消息费拒绝 取消消费的时候 调用cancel
             *String consumerTag: 消费者身份标识
             */
            @Override
            public void handle(String consumerTag) throws IOException {
                log.info("消费者:{},取消了消费",consumerTag);
            }
        };
        channel.basicConsume("queue01",true,
                deliver,
                cancel);
        Thread.sleep(10000000);
    }
}
